package com.xujian.cdc.sink.doris;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.xujian.cdc.convert.DateToStringConverter;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.JsonDebeziumSchemaSerializer;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.connect.json.JsonConverterConfig;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;

/**
 * @author star xu
 * @date 2023/3/28 16:24
 * @Description:schema自动变更
 * @Slogan: 致敬大师,致敬未来的自己
 */
public class Mysql2DorisCdcSchemaChange {

    public static void main(String[] args) throws Exception {

        Map<String, Object> customConverterConfigs = new HashMap<>();
        customConverterConfigs.put(JsonConverterConfig.DECIMAL_FORMAT_CONFIG, "numeric");
        JsonDebeziumDeserializationSchema schema =
                new JsonDebeziumDeserializationSchema(false, customConverterConfigs);

        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("127.0.0.1")
                .port(3306)
                .databaseList("flink-test") // set captured database
                .tableList("flink-test.table_hash") // set captured table
                .username("root")
                .password("qwer1234")
                .debeziumProperties(DateToStringConverter.DEFAULT_PROPS)
                .deserializer(schema)
                .serverTimeZone("Asia/Shanghai")
                .includeSchemaChanges(true) // converts SourceRecord to JSON String
                .build();
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Configuration configuration = new Configuration();
        configuration.setString(RestOptions.BIND_PORT, "8081");
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        env.setParallelism(1);
        // enable checkpoint

        env.enableCheckpointing(10000);
        Properties props = new Properties();
        props.setProperty("format", "json");
        props.setProperty("read_json_by_line", "true");
        props.setProperty("light_schema_change","true");
        DorisOptions dorisOptions = DorisOptions.builder()
                .setFenodes("192.168.16.112:8030")
                .setTableIdentifier("test.table_hash3")
                .setUsername("test")
                .setPassword("test@2023").build();

        DorisExecutionOptions.Builder  executionBuilder = DorisExecutionOptions.builder();
        executionBuilder.setLabelPrefix("label-doris" + UUID.randomUUID())
                .setStreamLoadProp(props).setDeletable(true);

        DorisSink.Builder<String> builder = DorisSink.builder();
        builder.setDorisReadOptions(DorisReadOptions.builder().build())
                .setDorisExecutionOptions(executionBuilder.build())
                .setDorisOptions(dorisOptions)
                .setSerializer(JsonDebeziumSchemaSerializer.builder().setDorisOptions(dorisOptions).build());

        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")//.print();
                .sinkTo(builder.build());

        env.execute("Print MySQL Snapshot + Binlog");
    }
}